package com.yifeng.repo.controller.kafka.configure;

import com.yifeng.repo.controller.kafka.message.consume.TagConsumerInterceptor;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.producer.Partitioner;
import org.springframework.boot.context.properties.ConfigurationProperties;

import java.util.List;
import java.util.Map;

/**
 * Created by daibing on 2024/4/29.
 */
@Data
@ConfigurationProperties(prefix = KafkaProperties.PREFIX)
public class KafkaProperties {

    public static final String PREFIX = "yfcloud.controller.kafka";

    /**
     * 是否启用生产管理器
     */
    private boolean enableProduceManager = false;

    /**
     * 是否启用Pull消费管理器
     */
    private boolean enablePullConsumeManager = true;

    /**
     * 是否启用API接口
     */
    private boolean enableMqApi = false;

    /**
     * api地址前缀
     */
    private String mqAdminPrefix = "/kafka";

    /**
     * 应用自己的topic（默认仅对应用自己的topic发布消息，对所有的topic都可以消费消息）
     */
    private String masterTopic;

    /**
     * 默认仅对应用自己的topic发布消息
     */
    private boolean masterTopicOnlyProduce = true;

    /**
     * 支持多个topic配置
     */
    private List<TopicProperties> topicPropertiesList;

    @Data
    public static class TopicProperties {
        /**
         * 基础配置
         */
        private String topic;
        private String namesrvAddr;
        private String username;
        private String password;
        private String consumerGroup;
        private Map<String, String> consumerGroupMapping;

        /**
         * 扩展配置：支持生产和消费是两套用户密码
         */
        private String producerUsername;
        private String producerPassword;
        private String consumerUsername;
        private String consumerPassword;

        /**
         * 生产配置
         * 特别注意：Kafka默认不支持幂等事务，需要关闭幂等特性。基于兼容要求，这里的enableIdempotence默认值不做调整。
         */
        private int sendMsgTimeout = 30 * 1000;
        private int retryTimes = 3;
        private int maxMessageSize = 1024 * 1024 * 4;
        private int batchSize = 16 * 1000;
        private int lingerMs = 1000;
        private boolean enableIdempotence = true;
        private Class<? extends Partitioner> customPartitionerClass;

        /**
         * Pull模式线程配置
         */
        private int pullBatchSize = 10;
        private Class<? extends ConsumerInterceptor<String, String>> consumerInterceptorClass = TagConsumerInterceptor.class;

    }

}
